// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: qiang

#include "coin2/exchange/apollox/order_v1/parser.h"

#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "coin/proto/coin_request.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/apollox/api_order/native_private_client.h"
#include "coin2/exchange/apollox/order_v1/error_code.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::apollox::order_v1 {

namespace {

using namespace coin2::exchange::apollox::order_v1::error_code;

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::CurrencyTransferElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin2::exchange::apollox::symbology::ApolloxCurrency;
using coin2::exchange::apollox::symbology::ApolloxProduct;
using coin2::exchange::base::order::IOrderContext;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;
inline int64_t GetTransferTimestamp(std::string time_str) {
  time_str.replace(10, 1, "T");
  return TimestampFromIso8601(time_str + ".000Z");
}

inline int64_t GetProcOrderId(const JsonValue& data) {
  int64_t proc_order_id = 0;
  if (data.HasMember("clientOrderId") && data["clientOrderId"].IsInt64()) {
    proc_order_id = stoll(data["clientOrderId"].GetString());
  }
  return proc_order_id;
}

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "BUY") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "SELL") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "LIMIT") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "MARKET") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

void SetOrderStatus(const std::string& status, ProductOrderElement* order) {
  if (status == "PENDING_CANCEL" || status == "REJECTED") {
    order->set_is_live(false);
  } else if (status == "NEW" || status == "CONFIRMED" || status == "PARTIALLY_FILLED") {
    order->set_is_live(true);
  } else if (status == "CANCELED" || status == "EXPIRED") {
    order->set_cancel_confirmed(true);
  } else if (status == "FILLED") {
    order->set_fully_filled(true);
  } else {
    LOG(ERROR) << "[ParseRestOrderInfo] Unknown order status! " << status;
  }
}

bool IsApolloxErrMessage(const JsonValue& doc) {
  return doc.HasMember("code") && doc.HasMember("msg");
}

bool IsAccountUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  return json.HasMember("e") &&
         std::string_view(json["e"].GetString()) == "outboundAccountPosition";
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  return json.HasMember("e") && std::string(json["e"].GetString()) == "executionReport";
}

// bool validate_raw_margin_entry(const JsonValue& doc) {
//   if (!doc.HasMember("asset") || !doc.HasMember("free") || !doc.HasMember("locked") ||
//       !doc.HasMember("netAsset")) {
//     return false;
//   } else {
//     return true;
//   }
// }

}  // namespace

void ApolloxParser::ParseRestSpotAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  // Linchuan: sometimes rest balance update slower than ws update about 10 sec,
  // makes balance "jump" back to wrong balance. solution: if ws balance is working,
  // use ws balance, if ws not update for 30 sec, use rest balance.
  if (GetCurrentTimestamp() - last_ws_balance_timestamp_ <= 30e9) {
    result_.set_type(MsgType::IGNORABLE);
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

bool ApolloxParser::ParseRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
    {
     "feeTier": 0,
     "canTrade": true,
     "canDeposit": true,
     "canWithdraw": true,
     "canBurnAsset": true,
     "updateTime": 0,
     "balances": [
      {
        "asset": "BTC",
        "free": "4723846.89208129",
        "locked": "0.00000000"
      },
      {
        "asset": "LTC",
        "free": "4763368.68006011",
        "locked": "0.00000000"
      }
      ]
    }
  */
  if (doc.IsObject() && IsApolloxErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }
  if (!doc.HasMember("balances") || !doc["balances"].IsArray()) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }

  for (const auto& data : doc["balances"].GetArray()) {
    if (!data.HasMember("asset") || !data.HasMember("free") || !data.HasMember("locked")) {
      continue;
    }
    auto current_free = stod(data["free"].GetString());
    auto current_locked = stod(data["locked"].GetString());
    CurrencyBalance balance;
    balance.set_currency_native(data["asset"].GetString());
    balance.set_available(current_free);
    balance.set_hold(current_locked);
    balance.set_total(current_free + current_locked);
    raw_balance_map->emplace(balance.currency_native(), balance);
  }

  return true;
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance ApolloxParser::ParseBalance(const JsonValue& doc) {
  RawBalanceMap raw_balance_map;
  auto success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Apollox");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
  is_account_balance_initialized_ = true;

  return acc_balance;
}

bool ApolloxParser::ParseRawBalanceFromWs(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "e":"outboundAccountInfo",
    "E":1594263002528,
    "m":2,
    "t":6,
    "b":0,
    "s":0,
    "T":true,
    "W":true,
    "D":true,
    "u":1594263002526,
    "B":[
        {
            "a":"BTC",
            "f":"1.35230710",
            "l":"0.00000000"
        },
        {
            "a":"LTC",
            "f":"0.00000000",
            "l":"0.00000000"
        },
        ...
    ]
  }

  From 2020-09-09
  {
    "e": "outboundAccountPosition", //Event type
    "E": 1564034571105,             //Event Time
    "u": 1564034571073,             //Time of last account update
    "B": [                          //Balances Array
      {
        "a": "ETH",                 //Asset
        "f": "10000.000000",        //Free
        "l": "0.000000"             //Locked
      }
    ]
  }
  */
  if (!(doc.HasMember("E"))) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }
  if (!(doc.HasMember("B") && doc["B"].IsArray())) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }

  for (const auto& currency_bal : doc["B"].GetArray()) {
    auto free = stod(currency_bal["f"].GetString());
    auto locked = stod(currency_bal["l"].GetString());
    CurrencyBalance balance;
    balance.set_currency_native(currency_bal["a"].GetString());
    balance.set_available(free);
    balance.set_hold(locked);
    balance.set_total(free + locked);
    raw_balance_map->emplace(balance.currency_native(), balance);
  }

  return true;
}

bool ApolloxParser::ParseWsBalance(const JsonValue& doc, AccountBalance* acc_balance) {
  acc_balance->Clear();

  RawBalanceMap raw_balance_map;
  bool success = ParseRawBalanceFromWs(doc, &raw_balance_map);
  if (!success) {
    return false;
  }

  // update account balance, just publish delta change from ws update
  acc_balance->set_exchange("Apollox");
  acc_balance->set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    if (raw_balance_map.count(native_currency) == 0) {
      continue;
    }
    auto* balance = acc_balance->add_each_balance();
    balance->MergeFrom(raw_balance_map.at(native_currency));
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }

  return true;
}

ProductOrderElement ApolloxParser::ParseRestOrderInfo(const JsonValue& data) {
  /*
    {
      "avgPrice": "0.00000",
      "clientOrderId": "abc",
      "cumQuote": "0",
      "executedQty": "0",
      "orderId": 1917641,
      "origQty": "0.40",
      "origType": "STOP",
      "price": "0",
      "side": "BUY",
      "status": "NEW",
      "stopPrice": "9300",
      "closePosition": false,
      "symbol": "BTCUSDT",
      "time": 1579276756075,
      "timeInForce": "GTC",
      "type": "STOP",
      "updateTime": 1579276756075
    }
  */

  ProductOrderElement order;
  std::string native_symbol = data["symbol"].GetString();
  if (!symcache_->has_product_native(native_symbol)) {
    return order;
  }
  std::string exchange_id = fmt::format("{}{}", native_symbol, data["orderId"].GetString());
  if (GetCurrentTimestamp() / 1e9 - stoll(data["time"].GetString()) / 1e3 > 60 * 60) {
    auto* oc = GetOcm()->FindByExchangeOrderId(exchange_id);
    if (oc != nullptr && !GetOcm()->IsInWorkingOrders(oc)) {
      // TODO(xguo): adjust query url not to list all previous orders. Otherwise too chatty.
      // LOG(INFO) << "Ignore binance long time order: " << exchange_id;
      return order;
    }
  }

  order.set_market_type(MarketType::Spot);
  order.set_exchange_type(ExchangeType::Apollox);
  order.set_api_version("v1");
  order.set_exchange_order_id(exchange_id);
  order.set_symbol(native_symbol);
  order.set_price(stod(data["price"].GetString()));
  order.set_qty(stod(data["origQty"].GetString()));
  order.set_proc_order_id(GetProcOrderId(data));
  order.set_side(GetOrderSide(data["side"].GetString()));
  order.set_order_type(GetOrderType(data["type"].GetString()));
  order.set_timestamp(GetCurrentTimestamp());
  SetOrderStatus(data["status"].GetString(), &order);
  return order;
}

void ApolloxParser::ParseRestOpenOrders() {
  is_query_historical_orders_ = false;
  ParseRestOrderInfoList();
}

void ApolloxParser::ParseRestAllOrders() {
  is_query_historical_orders_ = true;
  ParseRestOrderInfoList();
}

void ApolloxParser::ParseRestOrderInfoList() {
  /*
  [
    ...
  ]
  */

  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOrderInfoList] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsApolloxErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOrderInfoList] " << JsonToString(data);
    return;
  }
  if (!data.IsArray()) {
    LOG(ERROR) << "Unexpected order list format!" << JsonToString(data);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();

  bool product_order_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto new_order = ParseRestOrderInfo(info);

    // Ignore open orders from QueryHistoricalOrders
    // only trust open roders from QueryOpenOrders
    if (!(is_query_historical_orders_ && new_order.is_live()) && new_order.has_market_type()) {
      auto* order = product_order_bundle->add_each_order();
      order->MergeFrom(new_order);
      product_order_bundle_found = true;
    }
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

ProductOrderBundle ApolloxParser::ParseWsOrderInfoListV1(const JsonValue& data) {
  ProductOrderBundle product_order_bundle;

  auto* order = product_order_bundle.add_each_order();
  order->set_market_type(MarketType::Spot);
  order->set_exchange_type(ExchangeType::Apollox);
  order->set_api_version("v1");
  order->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
  order->set_symbol(data["s"].GetString());
  order->set_price(stod(data["p"].GetString()));
  order->set_qty(stod(data["q"].GetString()));
  if (std::string(data["C"].GetString()).empty()) {
    order->set_proc_order_id(stoll(data["c"].GetString()));
  } else {
    order->set_proc_order_id(stoll(data["C"].GetString()));
  }
  order->set_side(GetOrderSide(data["S"].GetString()));
  order->set_order_type(GetOrderType(data["o"].GetString()));
  SetOrderStatus(data["X"].GetString(), order);
  order->set_timestamp(GetCurrentTimestamp());
  return product_order_bundle;
}

ProductFillBundle ApolloxParser::ParseWsFillInfoListV1(const JsonValue& data) {
  ProductFillBundle product_fill_bundle;

  std::string status = data["x"].GetString();
  if (status == "TRADE" || status == "AGGTRADE") {
    auto* fill = product_fill_bundle.add_each_fill();
    fill->set_symbol(data["s"].GetString());
    fill->set_fill_timestamp(std::stoll(data["E"].GetString()) * 1000000LL);
    fill->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
    if (std::string(data["C"].GetString()).empty()) {
      fill->set_proc_order_id(stoll(data["c"].GetString()));
    } else {
      fill->set_proc_order_id(stoll(data["C"].GetString()));
    }
    fill->set_price(stod(data["L"].GetString()));
    fill->set_qty(stod(data["l"].GetString()));
    fill->set_fill_id(data["t"].GetString());
    if (data["m"].GetBool()) {
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (data.HasMember("N") && !data["N"].IsNull()) {
      const std::string fee_currency = data["N"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {
        fill->set_fee_currency(fee_currency);
        if (data.HasMember("n") && !data["n"].IsNull()) {
          const std::string fee = data["n"].GetString();
          if (!fee.empty()) {
            fill->set_fee(-stod(fee));
          }
        }
      }
    }
  }
  return product_fill_bundle;
}

void ApolloxParser::ParseRestFillInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestFillOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsApolloxErrMessage(data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] " << JsonToString(data);
    return;
  }
  if (!(data.IsArray() && data.Size() > 0)) {
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();

  bool product_fill_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto* fill = product_fill_bundle->add_each_fill();
    fill->set_symbol(info["symbol"].GetString());
    fill->set_fill_timestamp(std::stoll(info["time"].GetString()) * 1000000LL);
    fill->set_exchange_order_id(
        info["symbol"].GetString() + std::string(info["orderId"].GetString()));
    fill->set_price(stod(info["price"].GetString()));
    fill->set_qty(stod(info["qty"].GetString()));
    fill->set_fill_id(info["id"].GetString());
    if (info["maker"].GetBool()) {  //等待确认
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (info.HasMember("commissionAsset") && !info["commissionAsset"].IsNull()) {
      const std::string fee_currency = info["commissionAsset"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {  //这个是什么意思？？
        fill->set_fee_currency(fee_currency);
        if (info.HasMember("commission") && !info["commission"].IsNull()) {
          const std::string fee = info["commission"].GetString();
          if (!fee.empty()) {
            fill->set_fee(-stod(fee));
          }
        }
      }
    }
    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

void ApolloxParser::ParseSubmitResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_timestamp(GetCurrentTimestamp());

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[ApolloxSubmitOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  const auto http_code = http_context->res.result_int();
  if (http_code == 429 || http_code == 418) {
    LOG(ERROR) << "[ApolloxSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_exchange_error_msg(http_context->res.body().data());
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[ApolloxSubmitOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code =
        GetOrderErrorCode(json["code"], json["msg"], order_context->has_exchange_order_id());
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[ApolloxSubmitOrder] Error Response: \n" << http_context->res;
    }
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_exchange_order_id(
      json["symbol"].GetString() + std::string(json["orderId"].GetString()));
}

std::string ApolloxParser::ParseHeader(const std::shared_ptr<AsioHttpContext>& http_context) {
  const auto& header = http_context->res.base();
  nlohmann::json json = {{"x-mbx-order-count-1d", -1},
                         {"x-mbx-order-count-10s", -1},
                         {"x-mbx-order-count-1m", -1},
                         {"x-mbx-used-weight-1m", -1},
                         {"retry-after", -1},
                         {"bind_ip", http_context->bind_ip}};

  if (header.count("x-mbx-used-weight-1m") > 0) {
    json["x-mbx-used-weight-1m"] = std::stoi(header.at("x-mbx-used-weight-1m").data());
  }

  if (header.count("x-mbx-order-count-10s") > 0) {
    json["x-mbx-order-count-10s"] = std::stoi(header.at("x-mbx-order-count-10s").data());
  }

  if (header.count("x-mbx-order-count-1d") > 0) {
    json["x-mbx-order-count-1d"] = std::stoi(header.at("x-mbx-order-count-1d").data());
  }

  if (header.count("retry-after") > 0) {
    json["retry-after"] = std::stoi(header.at("retry-after").data());
  }
  return json.dump();
}

void ApolloxParser::ParseCancelResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[ApolloxCancelOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  const auto http_code = http_context->res.result_int();
  if (http_code == 429 || http_code == 418) {
    LOG(ERROR) << "[ApolloxCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_exchange_error_msg(http_context->res.body().data());
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[ApolloxCancelOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }
  if (json.HasMember("symbol") && json.HasMember("orderId")) {
    resp->set_exchange_order_id(
        json["symbol"].GetString() + std::string(json["orderId"].GetString()));
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code =
        GetOrderErrorCode(json["code"], json["msg"], order_context->has_exchange_order_id());
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[ApolloxCancelOrder] Error Response" << http_context->res;
      resp->set_success(false);
      return;
    }
  }

  resp->set_success(true);
  resp->set_cancel_confirmed(true);  // TODO 这里需要确认， 如果发送错误的订单会是什么样子的
}

void ApolloxParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (IsAccountUpdate(document)) {
    // make first rest balance result before ws balance result,
    // so SetAllZeroBalance won't take effect on partial symbol
    if (!is_account_balance_initialized_) {
      result_.set_type(MsgType::IGNORABLE);
      return;
    }
    if (!ParseWsBalance(document, result_.mutable_account_balance())) {
      result_.set_error_code(OrderErrorCode::INVALID_JSON);
      return;
    }
    result_.set_type(MsgType::ACCOUNT_BALANCE);
    last_ws_balance_timestamp_ = GetCurrentTimestamp();
  } else if (IsOrderUpdate(document)) {
    result_.fill_bundle_ = ParseWsFillInfoListV1(document);
    result_.order_bundle_ = ParseWsOrderInfoListV1(document);
    result_.set_type(MsgType::ORDER_STATUS);
    result_.set_type(MsgType::ORDER_FILL);
  }
}

void ApolloxParser::ParseListenKey() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseListenKey] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  // {"listenKey": "pqia91ma19a5s61cv6a81va65sdf19v8a65a1a5s61cv6a81va65sdf19v8a65a1"}
  LOG(INFO) << "Apollox/spot: " << JsonToString(document);
  if (document.HasMember("listenKey")) {
    std::string listen_key(document["listenKey"].GetString());
    if (mb_) {
      mb_->PublishEvent("apollox/listen_key", &listen_key);
    }
  } else {
    LOG(INFO) << "Failed to fetch listen key" << JsonToString(document);
  }
}

auto ApolloxParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg_->type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  std::string s = ParseHeader(msg.http_context);
  if (mb_) {
    mb_->PublishEvent("apollox/req_rate_limit", reinterpret_cast<void*>(&s));
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::LISTEN_KEY:
      ParseListenKey();
      break;
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      // pass
      break;
    case OrderSubTopicId::REST_ALL_ORDERS:
      ParseRestAllOrders();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpenOrders();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillInfoList();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestSpotAccountBalance();
      break;
    default: {
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }

  return result_;
}

}  // namespace coin2::exchange::apollox::order_v1
